package com.trust.pams.dao.service.impl;

import cn.hutool.http.HttpStatus;
import com.alibaba.fastjson.JSON;
import com.trust.pams.dao.service.EsPamsClientDao;
import com.trust.pams.dao.service.EsPamsPuoteIdDao;
import com.trust.pams.model.es.ClientEsSearch;
import com.trust.pams.model.es.QuoteEsSearch;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;

@Repository
public class EsPamsClientDaoImpl implements EsPamsClientDao {

    private static final Logger log = LoggerFactory.getLogger(EsPamsPuoteIdDaoImpl.class);


    @Resource
    private RestHighLevelClient restHighLevelClient;


    @Override
    public boolean batchAddDocument(List<ClientEsSearch> itemEsSearchList, String indexName) {
        BulkRequest bulkRequest = new BulkRequest();
        itemEsSearchList.stream().forEach(item->{
            bulkRequest.add(new IndexRequest(indexName + "__pams_client").source(JSON.toJSONString(item), XContentType.JSON));
        });

        try{
            BulkResponse responses = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (responses.status().getStatus()== HttpStatus.HTTP_OK){
                return true;
            }
        }catch (IOException e){
            log.error("batchIndex exception",e);
        }
        return false;
    }

    @Override
    public long dleteDocument(String id, String indexName) {
        //参数为索引名，可以不指定，可以一个，可以多个
        DeleteByQueryRequest request = new DeleteByQueryRequest(indexName + "__pams_client");
        // 更新时版本冲突
        request.setConflicts("proceed");
        // 设置查询条件，第一个参数是字段名，第二个参数是字段的值
        request.setQuery(new TermQueryBuilder("clientName", id));
        // 批次大小
        request.setBatchSize(1000);
        // 并行
        request.setSlices(2);
        // 使用滚动参数来控制“搜索上下文”存活的时间
        request.setScroll(TimeValue.timeValueMinutes(10));
        // 超时
        request.setTimeout(TimeValue.timeValueMinutes(2));
        // 刷新索引
        request.setRefresh(true);
        try {
            BulkByScrollResponse response = restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
            return response.getStatus().getUpdated();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return -1;
    }
}
